package spark.core.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;
import scala.actors.threadpool.Arrays;

/**
 * Created by Administrator on 2018/2/6/006.
 */
public class WordCountLocal {


    public static void main(String[] args) {
        //编写spark应用程序
        //第一步 创建 sparkConf对象，设置Spark应用的配置信息
        SparkConf conf = new SparkConf().setMaster("local").setAppName("WordCountLocal");
        //使用setMaster()可以设置spark应用程序要连接的Spark应用程序要连接的spark集群的Master节点的URL
        //第二步：创建JavaSparkContext对象
        //在Spark中，SparkContext，它的主要作用，包括初始化应用程序所需的一些核心组件。
        //调度器（DAGScheduler、TaskScheduler）,还回去到Spark Master节点上进行注册，等等
        //一句话，SparkContext是spark应用中，可以说是最重要的一个对象
        //但是呢，在spark中。编写不同类型的spark应用程序，使用的SparkContext是不同的，如果使用Scala
        JavaSparkContext sc = new JavaSparkContext(conf);
        //第三步：要针对输入源，创建一个初始的RDD
        //输入源中大数据会打散分配到Rdd的每一个partition中，从而形成一个初始的分布式数据集
        //我们这里吧，因为是本地测试，所以呢，就是针对本地文件
        //在JAVA中，用于根据文件类型的输入源创建RDD的方法，叫做textFile（）方法
        //在这里呢，RDD中，有元素这种概念。如果是HDFS或者是本地文件呢，创建RDD，每一个元素就相当于是文件里的一行
        JavaRDD<String>  lines = sc.textFile("datas/wordcount.txt");
        //第四步：对初始RDD进行TRANSFORMATION操作，也就是一些计算操作
        //通常操作会通过创建function，并配合RDD的map、flatmap等算子来执行function，通常，如果比较简单，则创建指定function的匿名内部类
       //但是如果function比较复杂，这回单独创建一个类，作为思想这个function接口的类
        //我们这里呢，输入肯定是String，因为一行一行的文本，输出，其实也是String。因为是每一行的文本
        //这里要先简要介绍flatmap算子的作用，其实就是，将RDD的一个元素，给拆分成一个或多个元素
        //先将每一行拆分成一个一个的单词
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            public Iterable<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" "));
            }
        });
        //接着，需要将每一个单词，映射为（单词，1）的这种格式
        //因为只要这样，后面才能根据单词作为KEY，来进行每个单词的出现次数的累加
        //mapToPair，其实就是将每个元素，映射为一个（v1,v2）这样的TUPLE2就是SCALA类型，包含了两个值
        //mapToPair这个算子，要求的是与pairFunction配合使用，第一个泛型参数代表了输入类型
        //第二个参数和第三个泛型参数，代表的输出的Tuple2的第一个值和第二个值的类型
       //JavaPairRDD的两个泛型参数分别代表了Tuple元素的第一个值和第二个值的类型
        JavaPairRDD<String,Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<String, Integer>(word ,1);
            }
        });
        //接着需要以单词作为KEY，统计每个单词出现的次数
        //这里使用reduceByKey这个算子。对每个key对应
        //比如JavaPairRDD中有几个元素，（hello,1）,（hello,1）,（hello,1）,(world,1)
        //reduce操作，相当于是把第一个值和第二个值进行计算，然后再将结果与第三个值进行计算
        //比如这里的hello，那么就相当于，首先是1+1=2.再将2+1=3
        //最后返回的JavaPairRDD中元素，也是TUPLE。但是第一个值就是每个KEY，第二个值就是key的value
        //reduce的结果，相当于就是每个单词出现的次数
        final JavaPairRDD<String,Integer> wordcounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        //到这里为止。，我们通过几个spark的算子操作，已经统计出了单词的次数。
        //但是，之前我们使用的flatmap,mapToPair，reduceByKey这种操作，都叫做transformation操作
        //一个spark应用中，可以使用一种叫做action的操作，。比如说foreach()，来触发程序的执行
        wordcounts.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            public void call(Tuple2<String, Integer> wordcount) throws Exception {
                System.out.println(wordcount._1()+" 出现了"+ wordcount._2() + " times.");
            }
        });

        sc.close();
    }
}
